package cn.bellychang.controller;

import cn.bellychang.deferredresult.DeferredResultHolder;
import cn.bellychang.deferredresult.MockQueue;
import cn.bellychang.model.ORDER_STATUS_ENUM;
import cn.bellychang.model.Order;
import cn.bellychang.response.ErrorCode;
import cn.bellychang.response.Response;
import cn.bellychang.service.OrderService;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.context.request.async.DeferredResult;
import org.springframework.web.context.request.async.WebAsyncTask;

import java.util.concurrent.*;
import java.util.function.Function;

import static cn.bellychang.controller.AsyncController.Delayer.delayer;

/**
 * @author ChangLiang
 * @date 2020/9/23
 */
@RestController
public class AsyncController {

    private Logger LOGGER = LoggerFactory.getLogger(getClass());

    @Autowired
    private MockQueue mockQueue;

    @Autowired
    private DeferredResultHolder deferredResultHolder;

    @Autowired
    private ObjectMapper objectMapper;

    @Autowired
    private OrderService orderService;

    @Autowired
    private ThreadPoolTaskExecutor callableAsyncExecutor;

    /**
     * 这个请求由于Future#get 所以仍旧是阻塞的
     *
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     */
    @GetMapping("/completableFuture/order")
    public Response orderCompletableFuture() throws InterruptedException, ExecutionException {
        LOGGER.info("tomcat主线程开始");
        Order order = new Order();
        order.setId(ThreadLocalRandom.current().nextLong(0, 99999));
        order.setStatus(ORDER_STATUS_ENUM.init.name());
        Response response = new Response();

        CompletableFuture<Response> completableFuture = CompletableFuture.supplyAsync(() -> {
            Response resp = new Response();
            ErrorCode errorCode = ErrorCode.OK;
            try {
                LOGGER.info("副线程开始");
                orderService.processOrder(order);
                LOGGER.info("副线程返回");
                resp.setData(order);
            } catch (Exception e) {
                errorCode = ErrorCode.ERROR;
            }
            resp.setErrorCode(errorCode.code);
            resp.setValue(errorCode.value);
            return resp;
        }).whenComplete((v, t) -> {
            LOGGER.info("副线程执行完成");
        });
        try {
            response = completableFuture.get(2, TimeUnit.SECONDS);
        } catch (TimeoutException e) {
            ErrorCode errorCode = ErrorCode.ERROR;
            response.setErrorCode(errorCode.code);
            response.setValue(errorCode.value);
        }
        LOGGER.info("tomcat主线程结束");
        return response;
    }

    @GetMapping("/completionStage/order")
    public CompletionStage<Response> orderCompletionStage() {
        LOGGER.info("tomcat主线程开始");
        Order order = new Order();
        order.setId(ThreadLocalRandom.current().nextLong(0, 99999));
        order.setStatus(ORDER_STATUS_ENUM.init.name());

        return within(CompletableFuture.supplyAsync(() -> {
            Response resp = new Response();
            ErrorCode errorCode = ErrorCode.OK;
            try {
                LOGGER.info("副线程开始");
                orderService.processOrderWithThreadSleep(order);
                LOGGER.info("副线程返回");
                resp.setData(order);
            } catch (Exception e) {
                errorCode = ErrorCode.ERROR;
            }
            resp.setErrorCode(errorCode.code);
            resp.setValue(errorCode.value);
            return resp;
        }), 2, TimeUnit.SECONDS)
                .exceptionally(throwable -> {
                    LOGGER.error("throwable:{}", throwable.getMessage());
                    Response resp = new Response();
                    resp.setErrorCode(ErrorCode.ERROR.code);
                    resp.setValue(ErrorCode.ERROR.value);
                    return resp;
                });
    }

    public static <T> CompletableFuture<T> within(CompletableFuture<T> future, long timeout, TimeUnit unit) {
        final CompletableFuture<T> timeoutFuture = timeoutAfter(timeout, unit);
        // 哪个先完成 就apply哪一个结果
        return future.applyToEither(timeoutFuture, Function.identity());
    }

    public static <T> CompletableFuture<T> timeoutAfter(long timeout, TimeUnit unit) {
        CompletableFuture<T> result = new CompletableFuture<T>();
        // timeout 时间后 抛出TimeoutException 类似于sentinel / watcher
        delayer.schedule(() -> result.completeExceptionally(new TimeoutException()), timeout, unit);
        return result;
    }

    /**
     * Singleton delay scheduler, used only for starting and * cancelling tasks.
     */
    static final class Delayer {
        static ScheduledFuture<?> delay(Runnable command, long delay,
                                        TimeUnit unit) {
            return delayer.schedule(command, delay, unit);
        }

        static final class DaemonThreadFactory implements ThreadFactory {
            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r);
                t.setDaemon(true);
                t.setName("CompletableFutureDelayScheduler");
                return t;
            }
        }

        static final ScheduledThreadPoolExecutor delayer;

        // 注意，这里使用一个线程就可以搞定 因为这个线程并不真的执行请求 而是仅仅抛出一个异常
        static {
            (delayer = new ScheduledThreadPoolExecutor(
                    1, new DaemonThreadFactory())).
                    setRemoveOnCancelPolicy(true);
        }
    }


    @GetMapping("/callable/order")
    public WebAsyncTask<Response> orderCallable() {

        LOGGER.info("tomcat主线程开始");
        Order order = new Order();
        order.setId(ThreadLocalRandom.current().nextLong(0, 99999));
        order.setStatus(ORDER_STATUS_ENUM.init.name());

        Response response = new Response();

        ///
        /**
         * cn.bellychang.interceptor.InterceptorRegistryConfig
         * configurer.setTaskExecutor(callableAsyncExecutor());
         * 选择是直接使用新建线程的方式 还是使用线程池的方式
         */
        Callable<Response> result = () -> {
            ErrorCode errorCode = ErrorCode.OK;
            try {
                LOGGER.info("副线程开始");
                orderService.processOrder(order);
                LOGGER.info("副线程返回");
                response.setData(order);
            } catch (Exception e) {
                errorCode = ErrorCode.ERROR;
            }
            response.setErrorCode(errorCode.code);
            response.setValue(errorCode.value);
            return response;
        };


        LOGGER.info("tomcat主线程结束");
        return new WebAsyncTask<>(2000, result);
    }

    @GetMapping("/deferredResult/order")
    public DeferredResult<Response> orderDeferredResult() {

        // 应用1的tomcat主线程来接收http请求 订单初始化
        LOGGER.info("tomcat主线程开始");
        Order order = new Order();
        order.setId(ThreadLocalRandom.current().nextLong(0, 99999));
        order.setStatus(ORDER_STATUS_ENUM.init.name());

        // 模拟将订单放到消息队列中
        mockQueue.receiveOrder(order);

        DeferredResult<Response> deferredResult = new DeferredResult<>(4000L);
        // 三种类型的回调
        deferredResult.onCompletion(() -> LOGGER.info("Processing complete"));

        // http code 408 request timeout
        /*deferredResult.onTimeout(() ->
                deferredResult
                        .setErrorResult(ResponseEntity.status(HttpStatus.REQUEST_TIMEOUT)
                                .body("Request timeout occurred")
                        ));*/

        /**
         * 自定义返回超时
         */
        deferredResult.onTimeout(() -> {
                    Response response = new Response();
                    response.setErrorCode(ErrorCode.REQUEST_TIMEOUT.code);
                    response.setValue(ErrorCode.REQUEST_TIMEOUT.value);
                    try {
                        String s = objectMapper.writeValueAsString(response);
                        deferredResult.setErrorResult(ResponseEntity.status(HttpStatus.OK)
                                .contentType(MediaType.APPLICATION_JSON)
                                .body(objectMapper.writeValueAsString(response)));
                    } catch (JsonProcessingException e) {
                        e.printStackTrace();
                    }
                }
        );

        deferredResult.onError(t -> LOGGER.error("Processing error, message: {}", t.getMessage()));

        // 把订单号作为key 放到deferredResultHolder中
        try {
            deferredResultHolder.getMap().put(order.getId(), deferredResult);
        } catch (Exception e) {
            e.printStackTrace();
        }
        LOGGER.info("tomcat主线程结束");
        return deferredResult;
    }
}
